package com.spark.sql

import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object Demo4DataSource {

  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .appName("source")
      .master("local")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()
    import org.apache.spark.sql.functions._
    import spark.implicits._

    /**
     * csv格式的数据
     * 选哟指定表结构，分隔符（默认时逗号），文件路径
     *
     */

    val csvDF: DataFrame = spark
      .read
      .format("csv")
      .schema("id STRING,name STRING,age INT,gender STRING, clazz STRING") //指定列名和类的类型
      .option("sep", ",")
      .load("data/students.txt")

    csvDF.show()

    /**
     * 将数据保持为csv格式
     * 按照班级分组，得到班级的人数,并将结果写入到指定目录
     */
    csvDF
      .groupBy($"clazz")
      .agg(count($"clazz") as "num")
      .write //保持数据
      .format("csv") //指定保持数据的格式
      .option("sep", ",")
      .mode(SaveMode.Overwrite)
      .save("data/clazz_num1")


    /**
     * json格式
     * spark 会自动将json中字段名和字段类型解析出来
     *
     * json格式比csv格式占用的空间更大，在大数据场景下不适用json
     *
     */

    //读取json格式的数据
    val jsonDF: DataFrame = spark.read
      .format("json")
      .load("data/students.json")

    jsonDF.show()
    //将数据保存为json格式
    jsonDF
      .groupBy($"gender")
      .agg(count($"gender") as "c")
      .write
      .format("json")
      .mode(SaveMode.Overwrite)
      .save("data/gender_num")

    /**
     *
     * parquet: 带表结构的压缩格式
     * 压缩：时间换空间
     * 压缩比取决于《信息熵》
     *
     */

    //1、将数据保存为parquet
    jsonDF
      .write
      .format("parquet")
      .mode(SaveMode.Overwrite)
      .save("data/students")


    /**
     * 读取parquet
     * parquet格式的数据自带了表结构，不需要手动指定
     *
     */

    val parquetDF: DataFrame = spark
      .read
      .format("parquet")
      .load("data/students")

    parquetDF.printSchema()
    parquetDF.show()


    /**
     * 读取JDBC中的数据
     *
     */

    val jdbcDF: DataFrame = spark
      .read
      .format("jdbc")
      .option("url", "jdbc:mysql://master:3306")
      .option("dbtable", "bigdata.students")
      .option("user", "root")
      .option("password", "123456")
      .load()

    jdbcDF.printSchema()
    jdbcDF.show()



  }

}
